package cn.etgps.etgps.utils;

import org.reactivestreams.Publisher;

import io.reactivex.Flowable;
import io.reactivex.FlowableTransformer;
import io.reactivex.Observable;
import io.reactivex.ObservableTransformer;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.disposables.CompositeDisposable;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Consumer;
import io.reactivex.schedulers.Schedulers;

/**
 * Created by zh on 16/5/6.
 * rx切换线程
 */
public class RxSchedulers {
	public static <T> FlowableTransformer<T, T> io_main_flowable() {
		return new FlowableTransformer<T, T>() {
			@Override
			public Publisher<T> apply(Flowable<T> tObservable) {
				return tObservable.subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread());
			}
		};
	}

	public static <T> ObservableTransformer<T, T> io_main_observalbe(final CompositeDisposable d) {
		return new ObservableTransformer<T, T>() {
			@Override
			public Observable<T> apply(Observable<T> tObservable) {
				return tObservable.subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread())
						.doOnSubscribe(new Consumer<Disposable>() {
							@Override
							public void accept(Disposable disposable) throws Exception {
								if (d != null) {
									d.add(disposable);
								}
							}
						});
			}
		};
	}
}
